package spark.sql

import org.apache.spark.sql.SaveMode
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created by ibf on 2018/2/10.
 */
object ReadCSVSparkSQL {
  def main(args: Array[String]) {

    //1、构建spark上下文
    val conf = new SparkConf()

      /**
       * 当执行代码的时候，sparksql会自动帮你分为200分区！
       * 当数据量大的时候，200个不太够
       * 当数据量小的时候（测试），200非常耽误时间
       * 所以通过这个spark.sql.shuffle.partitions参数来手动设定shuffle的时候，分区数量
       *
       * spark.sql.autoBroadcastJoinThreshold  10M 自动广播变量
       * join时候可以去把reduce join变成map join
       * map join适用的场景 大表join小表，差距在100多倍
       * 在spark当中，会将小表自动作为广播变量发送到每一个executor上
       */
      .set("spark.sql.shuffle.partitions","5")
      .setMaster("local[*]")
      .setAppName("ReadCSVSparkSQL")
    val sc = SparkContext.getOrCreate(conf)
    val sqlContext = new HiveContext(sc)

    //2、生成DataFrame或者临时表（tmptable）
    //创建schema信息
    val schema = StructType(Array(
      StructField("id",IntegerType),
      StructField("lat",StringType,nullable = false),
      StructField("lon",StringType,nullable = false),
      StructField("time",StringType)
    ))
    //数据路径
    val readDataPath = "datas/taxi.csv"
    sqlContext.read
      //这个数据格式可以给定短名称
      //也可以给定全名（包名+短名称）
      //全名：com.databricks.spark.csv，但是一般情况下，都是使用短名称
      .format("csv")
      //因为我们的数据中第一行并不是列名，所以指定false
      .option("header","false")
      .schema(schema)
      .load(readDataPath)
      .registerTempTable("tmp_taxi")

    //3、获取给定格式的数据数据，并将其注册成临时表
    sqlContext.sql(
      """
        |SELECT
        |id,substring(time,0,2) as hour
        |FROM tmp_taxi
      """.stripMargin)
      .registerTempTable("tmp_id_hour")

    //4、计算各个小时的各个出租车的载客次数
    //思考：group by 什么字段
    //答案：group by hour,id
    /**
     * count(*)和count(1)的区别
     * 没有任何区别，得到的数据结果是一模一样
     * 可以测试：
     *  count(*)和count(1)哪个运行快，就使用哪个
     *  如果都差不多，个人偏向 *
     */
    sqlContext.sql(
      """
        |SELECT
        |id,hour,count(*) as count
        |FROM tmp_id_hour
        |GROUP BY id,hour
      """.stripMargin)
    .registerTempTable("tmp_id_hour_count")

    //5、计算各个小时段中，载客次数最多的前五个出租车的载客情况和详细信息
    sqlContext.sql(
      """
        |SELECT
        |id,hour,count,
        |ROW_NUMBER() OVER(PARTITION BY hour ORDER BY count DESC) as rnk
        |FROM tmp_id_hour_count
      """.stripMargin)
    .registerTempTable("tmp_id_hour_count_rnk")

    val resultDataFrame = sqlContext.sql(
      """
        |SELECT
        |id,hour,count,rnk
        |FROM tmp_id_hour_count_rnk
        |WHERE rnk <= 5
      """.stripMargin)

    resultDataFrame.cache()

    resultDataFrame.show()
    resultDataFrame
                    .write
                   .format("csv")
                   .option("header","true")
                   .partitionBy("hour")
                   .mode(SaveMode.Overwrite)
                   .save(s"result/sparksql_csv_${System.currentTimeMillis()}")



    resultDataFrame.unpersist()

    Thread.sleep(1000000l)

  }
}
